package taillog

import (
    "LogAgent/kafka"
    "fmt"
    "github.com/hpcloud/tail"
)

type TailTask struct {
    Path  string
    Topic string
    instance *tail.Tail
    signal  chan string
}

func NewTailTask(topic string,path string) (tailObj *TailTask) {
    tailObj=&TailTask{
        Topic:topic,
        Path: path,
        signal: make(chan string),
    }

    tailObj.init()
    return
}


func (t *TailTask) init() (err error) {
    config := tail.Config{
        ReOpen:    true,
        Follow:    true,
        Location:  &tail.SeekInfo{Offset: 0, Whence: 2},
        MustExist: false,
        Poll:      true}
    t.instance, err = tail.TailFile(t.Path, config)
    if err != nil {
        fmt.Println("tail file failed, err:", err)
        return
    }

    go t.run()
    return
}

func (t *TailTask) run() {
tailLoop:
    for  {
        select {
        case lines:=<- t.instance.Lines:
            kafka.SendToLogDataChan(t.Topic,lines.Text)
        case <- t.signal:
            break tailLoop
        }
    }
}

func (t *TailTask) cancel()  {
     t.signal <- "1"
}

